{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(1)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "from pyflink.table.catalog import CatalogDatabase\n",
    "from pyflink.table.catalog import ObjectPath\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"db\" + os.sep\n",
    "\n",
    "ALINK_PLUGIN_DIR = \"/Users/yangxu/Downloads/alink_plugin\"\n",
    "\n",
    "IRIS_URL = \"http://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data\"\n",
    "IRIS_SCHEMA_STR = \"sepal_length double, sepal_width double, petal_length double, petal_width double, category string\"\n",
    "\n",
    "DB_NAME = \"test_db\"\n",
    "BATCH_TABLE_NAME = \"batch_table\"\n",
    "STREAM_TABLE_NAME = \"stream_table\"\n",
    "\n",
    "HIVE_VERSION = \"2.3.4\"\n",
    "HIVE_CONF_DIR = None\n",
    "\n",
    "DERBY_VERSION = \"10.6.1.0\"\n",
    "DERBY_DIR = \"derby\"\n",
    "\n",
    "MYSQL_VERSION = \"5.1.27\"\n",
    "MYSQL_URL = None\n",
    "MYSQL_PORT = None\n",
    "MYSQL_USER_NAME = None\n",
    "MYSQL_PASSWORD = None\n",
    "\n",
    "\n",
    "if None!=ALINK_PLUGIN_DIR :\n",
    "    AlinkGlobalConfiguration.setPluginDir(ALINK_PLUGIN_DIR)\n",
    "    \n",
    "    AlinkGlobalConfiguration.setPrintProcessInfo(True)\n",
    "    DOWNLOADER = AlinkGlobalConfiguration.getPluginDownloader()\n",
    "    \n",
    "    if None!=HIVE_CONF_DIR :\n",
    "        DOWNLOADER.downloadPlugin(\"hive\", HIVE_VERSION)\n",
    "    \n",
    "    if None!=DERBY_DIR :\n",
    "        DOWNLOADER.downloadPlugin(\"derby\", DERBY_VERSION)\n",
    "\n",
    "    if None!=MYSQL_URL :\n",
    "        DOWNLOADER.downloadPlugin(\"mysql\", MYSQL_VERSION)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2\n",
    "\n",
    "if None!=ALINK_PLUGIN_DIR and None!=HIVE_CONF_DIR :\n",
    "    hive = HiveCatalog(\"hive_catalog\", None, HIVE_VERSION, HIVE_CONF_DIR)\n",
    "    hive.open()\n",
    "\n",
    "    gateway = get_java_gateway()\n",
    "    database_properties = gateway.jvm.java.util.HashMap()\n",
    "    database_comment = None\n",
    "    j_database = gateway.jvm.org.apache.flink.table.catalog.CatalogDatabaseImpl(\n",
    "        database_properties, database_comment)\n",
    "    hive.create_database(DB_NAME, CatalogDatabase(j_database), True)\n",
    "\n",
    "    hive.drop_table(ObjectPath(DB_NAME, BATCH_TABLE_NAME), True)\n",
    "    hive.drop_table(ObjectPath(DB_NAME, STREAM_TABLE_NAME), True)\n",
    "\n",
    "    # batch sink\n",
    "    CsvSourceBatchOp()\\\n",
    "        .setFilePath(IRIS_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .lazyPrintStatistics(\"< origin data >\")\\\n",
    "        .link(\n",
    "            CatalogSinkBatchOp().setCatalogObject(CatalogObject(hive, ObjectPath(DB_NAME, BATCH_TABLE_NAME)))\n",
    "        )\n",
    "\n",
    "    BatchOperator.execute()\n",
    "\n",
    "    # stream sink\n",
    "    CsvSourceStreamOp()\\\n",
    "        .setFilePath(IRIS_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .link(\n",
    "            CatalogSinkStreamOp()\\\n",
    "                .setCatalogObject(CatalogObject(hive, ObjectPath(DB_NAME, STREAM_TABLE_NAME)))\n",
    "    )\n",
    "\n",
    "    StreamOperator.execute()\n",
    "\n",
    "    # batch source\n",
    "    CatalogSourceBatchOp()\\\n",
    "        .setCatalogObject(CatalogObject(hive, ObjectPath(DB_NAME, BATCH_TABLE_NAME)))\\\n",
    "        .lazyPrintStatistics(\"< batch catalog source >\")\n",
    "\n",
    "    BatchOperator.execute();\n",
    "\n",
    "    # stream source\n",
    "    CatalogSourceStreamOp()\\\n",
    "        .setCatalogObject(CatalogObject(hive, ObjectPath(DB_NAME, STREAM_TABLE_NAME)))\\\n",
    "        .sample(0.02)\\\n",
    "        .print()\n",
    "\n",
    "    StreamOperator.execute()\n",
    "\n",
    "    # meta operation\n",
    "\n",
    "    print(\"< tables before drop >\")\n",
    "    print(hive.list_tables(DB_NAME))\n",
    "\n",
    "    if hive.table_exists(ObjectPath(DB_NAME, BATCH_TABLE_NAME)) :\n",
    "        hive.drop_table(ObjectPath(DB_NAME, BATCH_TABLE_NAME), False)\n",
    "    hive.drop_table(ObjectPath(DB_NAME, STREAM_TABLE_NAME), True)\n",
    "\n",
    "    print(\"< tables after drop >\")\n",
    "    print(hive.list_tables(DB_NAME))\n",
    "\n",
    "    hive.drop_database(DB_NAME, True)\n",
    "\n",
    "    hive.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_3\n",
    "\n",
    "if None!=ALINK_PLUGIN_DIR and None!=DERBY_DIR :\n",
    "    derby = DerbyCatalog(\"derby_catalog\", None, DERBY_VERSION, DATA_DIR + DERBY_DIR)\n",
    "\n",
    "    derby.open()\n",
    "\n",
    "    derby.create_database(DB_NAME, CatalogDatabase(None), True)\n",
    "\n",
    "    derby.drop_table(ObjectPath(DB_NAME, BATCH_TABLE_NAME), True)\n",
    "    derby.drop_table(ObjectPath(DB_NAME, STREAM_TABLE_NAME), True)\n",
    "\n",
    "    # batch sink\n",
    "    CsvSourceBatchOp()\\\n",
    "        .setFilePath(IRIS_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .lazyPrintStatistics(\"< origin data >\")\\\n",
    "        .link(\n",
    "            CatalogSinkBatchOp().setCatalogObject(CatalogObject(derby, ObjectPath(DB_NAME, BATCH_TABLE_NAME)))\n",
    "        )\n",
    "\n",
    "    BatchOperator.execute()\n",
    "\n",
    "    # stream sink\n",
    "    CsvSourceStreamOp()\\\n",
    "        .setFilePath(IRIS_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .link(\n",
    "            CatalogSinkStreamOp()\\\n",
    "                .setCatalogObject(CatalogObject(derby, ObjectPath(DB_NAME, STREAM_TABLE_NAME)))\n",
    "    )\n",
    "\n",
    "    StreamOperator.execute()\n",
    "\n",
    "    # batch source\n",
    "    CatalogSourceBatchOp()\\\n",
    "        .setCatalogObject(CatalogObject(derby, ObjectPath(DB_NAME, BATCH_TABLE_NAME)))\\\n",
    "        .lazyPrintStatistics(\"< batch catalog source >\")\n",
    "\n",
    "    BatchOperator.execute();\n",
    "\n",
    "    # stream source\n",
    "    CatalogSourceStreamOp()\\\n",
    "        .setCatalogObject(CatalogObject(derby, ObjectPath(DB_NAME, STREAM_TABLE_NAME)))\\\n",
    "        .sample(0.02)\\\n",
    "        .print()\n",
    "\n",
    "    StreamOperator.execute()\n",
    "\n",
    "    # meta operation\n",
    "\n",
    "    print(\"< tables before drop >\")\n",
    "    print(derby.list_tables(DB_NAME))\n",
    "\n",
    "    if derby.table_exists(ObjectPath(DB_NAME, BATCH_TABLE_NAME)) :\n",
    "        derby.drop_table(ObjectPath(DB_NAME, BATCH_TABLE_NAME), False)\n",
    "    derby.drop_table(ObjectPath(DB_NAME, STREAM_TABLE_NAME), True)\n",
    "\n",
    "    print(\"< tables after drop >\")\n",
    "    print(derby.list_tables(DB_NAME))\n",
    "\n",
    "    derby.drop_database(DB_NAME, True)\n",
    "\n",
    "    derby.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_3\n",
    "\n",
    "if None!=ALINK_PLUGIN_DIR and None!=MYSQL_URL :\n",
    "    mysql = MySqlCatalog(\"mysql_catalog\", \"mysql\", MYSQL_VERSION, MYSQL_URL, MYSQL_PORT, MYSQL_USER_NAME, MYSQL_PASSWORD)\n",
    "\n",
    "    mysql.open()\n",
    "\n",
    "    mysql.create_database(DB_NAME, CatalogDatabase(None), True)\n",
    "\n",
    "    mysql.drop_table(ObjectPath(DB_NAME, BATCH_TABLE_NAME), True)\n",
    "    mysql.drop_table(ObjectPath(DB_NAME, STREAM_TABLE_NAME), True)\n",
    "\n",
    "    # batch sink\n",
    "    CsvSourceBatchOp()\\\n",
    "        .setFilePath(IRIS_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .lazyPrintStatistics(\"< origin data >\")\\\n",
    "        .link(\n",
    "            CatalogSinkBatchOp().setCatalogObject(CatalogObject(mysql, ObjectPath(DB_NAME, BATCH_TABLE_NAME)))\n",
    "        )\n",
    "\n",
    "    BatchOperator.execute()\n",
    "\n",
    "    # stream sink\n",
    "    CsvSourceStreamOp()\\\n",
    "        .setFilePath(IRIS_URL)\\\n",
    "        .setSchemaStr(IRIS_SCHEMA_STR)\\\n",
    "        .link(\n",
    "            CatalogSinkStreamOp()\\\n",
    "                .setCatalogObject(CatalogObject(mysql, ObjectPath(DB_NAME, STREAM_TABLE_NAME)))\n",
    "    )\n",
    "\n",
    "    StreamOperator.execute()\n",
    "\n",
    "    # batch source\n",
    "    CatalogSourceBatchOp()\\\n",
    "        .setCatalogObject(CatalogObject(mysql, ObjectPath(DB_NAME, BATCH_TABLE_NAME)))\\\n",
    "        .lazyPrintStatistics(\"< batch catalog source >\")\n",
    "\n",
    "    BatchOperator.execute();\n",
    "\n",
    "    # stream source\n",
    "    CatalogSourceStreamOp()\\\n",
    "        .setCatalogObject(CatalogObject(mysql, ObjectPath(DB_NAME, STREAM_TABLE_NAME)))\\\n",
    "        .sample(0.02)\\\n",
    "        .print()\n",
    "\n",
    "    StreamOperator.execute()\n",
    "\n",
    "    # meta operation\n",
    "\n",
    "    print(\"< tables before drop >\")\n",
    "    print(mysql.list_tables(DB_NAME))\n",
    "\n",
    "    if mysql.table_exists(ObjectPath(DB_NAME, BATCH_TABLE_NAME)) :\n",
    "        mysql.drop_table(ObjectPath(DB_NAME, BATCH_TABLE_NAME), False)\n",
    "    mysql.drop_table(ObjectPath(DB_NAME, STREAM_TABLE_NAME), True)\n",
    "\n",
    "    print(\"< tables after drop >\")\n",
    "    print(mysql.list_tables(DB_NAME))\n",
    "\n",
    "    mysql.drop_database(DB_NAME, True)\n",
    "\n",
    "    mysql.close()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
